package com.lry.basic.redis;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import io.lettuce.core.ScriptOutputType;
import org.apache.poi.util.StringUtil;
import redis.clients.jedis.Jedis;

import java.lang.reflect.Type;
import java.util.Arrays;
import java.util.Collections;
import java.util.Set;
import java.util.UUID;

public class RedisDelayQueue<T> {

    private static class TaskItem<T>{
        public String id;
        public T msg;
    }

    private Type TaskType = new TypeReference<TaskItem<T>>() {}.getType();

    private Jedis jedis;
    private String queueKey;
    private long delayMillis;//延迟多少ms

    public RedisDelayQueue(Jedis jedis, String queueKey, long delayMillis) {
        this.jedis = jedis;
        this.queueKey = queueKey;
        this.delayMillis = delayMillis;
    }

    public void delay(T msg){
        TaskItem task = new TaskItem();
        task.id = UUID.randomUUID().toString(); // 分配唯一的 uuid
        task.msg = msg;
        String json = JSON.toJSONString(task);
        //key,score, member
        jedis.zadd(queueKey,System.currentTimeMillis()+delayMillis,json);
    }

    public void handleMsg(T msg) {
        System.out.println(msg);
    }

    public void loop1(){
        //返回中断标识，并清除中断
        while(!Thread.interrupted()){

            Set<String> members = jedis.zrangeByScore(queueKey,0,System.currentTimeMillis(),0,1);

            if (members.isEmpty()) {
                try {
                    System.out.println("没拿到");
                    Thread.sleep(500); // 歇会继续
                } catch (InterruptedException e) {
                    break;
                }
                continue;
            }

            String json  = members.iterator().next();
            if(jedis.zrem(queueKey,json)>0){
                TaskItem<T> task = JSON.parseObject(json, TaskType); // fast
                this.handleMsg(task.msg);
            }
        }
    }

    private String r(String key){
        String luaScript = "local resultArray = redis.call('zrangebyscore', KEYS[1], 0, ARGV[1], 'limit' , 0, 1)\n" +
                "if #resultArray > 0 then\n" +
                "    if redis.call('zrem', KEYS[1], resultArray[1]) > 0 then\n" +
                "        return resultArray[1]\n" +
                "    else\n" +
                "        return ''\n" +
                "    end\n" +
                "else\n" +
                "    return ''\n" +
                "end";
        String json = (String)jedis.eval(luaScript, 1, key, String.valueOf(System.currentTimeMillis()));
        return json;
    }

    public String r1(String key){
        String json = (String) jedis.evalsha("3ed90a7ba9f3b53a0a90edea825112f3a850b120", Collections.singletonList(queueKey), Collections.singletonList(""+System.currentTimeMillis()));
        return json;
    }


    public void loop(){
        //返回中断标识，并清除中断
        while(!Thread.interrupted()){
            //zrangebyscore 快log(n), 基于内存，持久化机制
            //没有重试机制，没有ack机制

            String json = r(queueKey);
            //没拿到消息
            if("".equals(json)){
                try {
                    System.out.println("没拿到");
                    Thread.sleep(500); // 歇会继续
                } catch (InterruptedException e) {
                    break;
                }
                continue;
            }

            TaskItem<T> task = JSON.parseObject(json, TaskType); // fast
            this.handleMsg(task.msg);
        }
    }

    public static void main(String[] args) {
        Jedis jedis = RedisPool.getJedis();

        RedisDelayQueue queue = new RedisDelayQueue<>(jedis, "q-demo",5000);

        Thread producer = new Thread() {
            public void run() {
                for (int i = 0; i < 10; i++) {
                    queue.delay("asdf" + i);
                }
            }
        };

        Thread consumer = new Thread() {
            public void run() {
                queue.loop();
            }
        };

        producer.start();
        consumer.start();

        try {
            //主线程必须等producer执行完毕才能往下走
            producer.join();
            //到这里producer执行完毕了
            Thread.sleep(6000);
            consumer.interrupt();
            //主线程必须等consumer执行完毕才能往下走
            consumer.join();
        } catch (InterruptedException e) {
        }

        RedisPool.release(jedis);
    }
}
